﻿using System;
using System.Collections.Generic;
using System.Data;
using System.Data.SqlClient;
using System.IO;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading;
using System.Transactions;
using CoreLab.Oracle;
using Common.Utilities;
using Common.Utilities.Dispatcher;
using IsolationLevel=System.Data.IsolationLevel;
using Workflows.Components.Entities.Relations.InferredAssociations;
using Workflows.Components.Entities.Distances;

namespace Workflows.Components.Entities.StreamCollapse
{
    public class StreamingCollapseWorker:IWorker
    {
        #region settings

        private const int CountPerCollapseSplit = 100;
        #endregion

        #region fields
        private Thread _WorkerThread;
        private StreamingCollapseInstruction _Instruction;
        private StreamingCollapseOutput _Output;
        private Guid _JobTicketID;
        private RunContext _Context;
        private string _WorkflowName;
        private string _ComponentName;

        /// <summary>
        /// 
        /// </summary>
        public string ComponentName
        {
            get { return _ComponentName; }
        }

        /// <summary>
        /// 
        /// </summary>
        public string WorkflowName
        {
            get { return _WorkflowName; }
        }
        // global
        // private List<int> _CollapsedEntityIDs;
        private Dictionary<string, IAttributeDistanceCalcutor> _AttributeComparers;
        #endregion

        #region IWorker Members

        public Thread WorkerThread
        {
            get { return _WorkerThread; }
        }

        public InstructionBase Instruction
        {
            get { return _Instruction; }
        }

        public ResultBase Output
        {
            get { return _Output; }
        }

        public Type ComponentType
        {
            get { return typeof(StreamingCollapseComponent); }
        }

        public RunContext Context
        {
            get { return _Context; }
            set { _Context = value; }
        }

        public void SetInstruction(Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._JobTicketID = ticketGuid;
            this._Instruction = (StreamingCollapseInstruction)request;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }

        public void SetInstruction(RunContext context, Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._JobTicketID = ticketGuid;
            this._Instruction = (StreamingCollapseInstruction)Instruction;
            this._Context = context;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }

        public void DoWork()
        {
            if (this.WorkerStartedEvent != null)
            {
                this.WorkerStartedEvent(this._JobTicketID, this, this.GetType(), this._Instruction);
            }
            this._WorkerThread = Thread.CurrentThread;
            this.CollapseEntitiesInStreamingMode();
        }


        public void DoWorkAsync()
        {
            if (this.WorkerStartedEvent != null)
            {
                this.WorkerStartedEvent(this._JobTicketID, this, this.GetType(), this._Instruction);
            }
            this._WorkerThread = new Thread(new ThreadStart(this.CollapseEntitiesInStreamingMode));
            this._WorkerThread.Start();
        }

        public void CancelWork()
        {
            if (this._WorkerThread != null && this._WorkerThread.IsAlive)
            {
                this._WorkerThread.Abort();
                if (this.WorkerCancelledEvent != null)
                {
                    this.WorkerCancelledEvent(this._JobTicketID, this.GetType(), this._Instruction);
                }
            }

        }

        public event WorkerStarted WorkerStartedEvent;
        public event WorkerFinished WorkerFinishedEvent;
        public event WorkerHasError WorkerErredEvent;
        public event WorkerCancelled WorkerCancelledEvent;
        public event WorkerStatusChanged WorkerStatusChangedEvent;

        #endregion

        #region work
        private void CollapseEntitiesInStreamingMode()
        {
            IDbConnection srcConn1 = null;
            IDbConnection srcConn2 = null;
            if (this._Instruction.DbProvider == DataProviderType.MSSQL)
            {
                srcConn1 = new SqlConnection(this._Instruction.ConnStr);
                srcConn2 = new SqlConnection(this._Instruction.ConnStr);
            }
            else
            {
                srcConn1 = new OracleConnection(this._Instruction.ConnStr);
                srcConn2 = new OracleConnection(this._Instruction.ConnStr);
            }
            SqlConnection tgtConn = new SqlConnection(this._Instruction.TgtConnStr);

            Dictionary<string, SqlDbType> srcFieldDbTypes = new Dictionary<string, SqlDbType>();
            TableColumn[] srcCols =
                DbUtil.GetTableColumns(
                    this._Instruction.DbProvider, this._Instruction.ConnStr, 
                    this._Instruction.SelectSql);
            SqlDbType entityIDDbType = SqlDbType.Int;
            foreach(TableColumn col in srcCols)
            {
                srcFieldDbTypes.Add(col.ColumnName, col.DbType);
                if(col.ColumnName.ToLower()==this._Instruction.IdAttributeName.ToLower())
                    entityIDDbType = col.DbType;
            }
            int totalEntityCount =
                DbUtil.GetTableRecordCount(
                    this._Instruction.DbProvider,
                    this._Instruction.ConnStr,
                    this._Instruction.SelectSql);

            // stored proc 
            TableColumn[] tgtCols =
                DbUtil.GetTableColumns(
                    DataProviderType.MSSQL,
                    this._Instruction.TgtConnStr, 
                    string.Empty, 
                    this._Instruction.CollapseInstanceTable);
            Dictionary<string, TableColumn> srcFieldTgtColMappings =
                new Dictionary<string, TableColumn>();
            foreach (TableColumn tgtCol in tgtCols)
            {
                if (tgtCol.ColumnName.ToLower() == this._Instruction.CollapsedIdAttributeName.ToLower())
                {
                    srcFieldTgtColMappings.Add(this._Instruction.CollapsedIdAttributeName, tgtCol);
                }
                foreach (string srcField in this._Instruction.CollapseInstanceFieldMappings.Keys)
                {
                    if (this._Instruction.CollapseInstanceFieldMappings[srcField] == tgtCol.ColumnName && 
                        srcField.ToLower()!=this._Instruction.CollapsedIdAttributeName.ToLower())
                    {
                        srcFieldTgtColMappings.Add(srcField, tgtCol);
                        break;
                    }
                }
            }

            // create stored procedure for insert into instance table
            string spNameInsertInstance = "bulkInsert_" + this._Instruction.CollapseInstanceTable + "_SP";
            TableColumn[] cols = new TableColumn[srcFieldTgtColMappings.Count];
            srcFieldTgtColMappings.Values.CopyTo(cols, 0);
            string spBody =
                StoredProcUtil.CreateStoredProcedureStatementForInsert(
                    DataProviderType.MSSQL, spNameInsertInstance, string.Empty,
                    this._Instruction.CollapseInstanceTable, cols);
            DbUtil.NewStoredProc(this._Instruction.TgtConnStr, spNameInsertInstance, spBody);
            Dictionary<string, SqlParameter> srcFieldParameterMappings =
                new Dictionary<string, SqlParameter>();
            if(srcFieldTgtColMappings.ContainsKey(this._Instruction.CollapsedIdAttributeName))
            {
                TableColumn tgtCol = srcFieldTgtColMappings[this._Instruction.CollapsedIdAttributeName];
                SqlParameter parameter = new SqlParameter("@" + tgtCol.ColumnName, DBNull.Value);
                srcFieldParameterMappings.Add(this._Instruction.CollapsedIdAttributeName, parameter);
            }
            foreach (string srcField in srcFieldTgtColMappings.Keys)
            {
                if (srcField.ToLower() != this._Instruction.CollapsedIdAttributeName.ToLower())
                {
                    TableColumn tgtCol = srcFieldTgtColMappings[srcField];
                    SqlParameter parameter = new SqlParameter("@" + tgtCol.ColumnName, DBNull.Value);
                    srcFieldParameterMappings.Add(srcField, parameter);
                }
            }

            // create stored procedure for insert into mapping table
            string spNameInsertMapping = "bulkInsert_" + this._Instruction.CollapseMappingTable + "_SP";
            TableColumn[] cols2 = DbUtil.GetTableColumns(DataProviderType.MSSQL,
                                                         this._Instruction.TgtConnStr, string.Empty,
                                                         this._Instruction.CollapseMappingTable);
            List<TableColumn> tgtCols2=new List<TableColumn>();
            Dictionary<string, SqlParameter> mappingTableSqlParameters = 
                new Dictionary<string, SqlParameter>();

            foreach(TableColumn col in cols2)
            {
                if(col.ColumnName==this._Instruction.CollapseMappingCollapseIDField)
                {
                    tgtCols2.Add(col);
                    mappingTableSqlParameters.Add(col.ColumnName, new SqlParameter("@" + col.ColumnName, DBNull.Value));
                }
                else if(col.ColumnName==this._Instruction.CollapseMappingEntityIDField)
                {
                    tgtCols2.Add(col);
                    mappingTableSqlParameters.Add(col.ColumnName, new SqlParameter("@" + col.ColumnName, DBNull.Value));
                }
                else if (col.ColumnName == this._Instruction.CollapseMappingScoreField)
                {
                    tgtCols2.Add(col);
                    mappingTableSqlParameters.Add(col.ColumnName, new SqlParameter("@" + col.ColumnName, DBNull.Value));
                }
                else if (col.ColumnName == this._Instruction.CollapseMappingTraceField)
                {
                    tgtCols2.Add(col);
                    mappingTableSqlParameters.Add(col.ColumnName, new SqlParameter("@" + col.ColumnName, DBNull.Value));
                }
                else if(this._Instruction.MappingTableFieldMappings !=null && this._Instruction.MappingTableFieldMappings.Count>0)
                {
                    foreach(string srcFieldName in this._Instruction.MappingTableFieldMappings.Keys)
                    {
                        if(this._Instruction.MappingTableFieldMappings[srcFieldName].ToLower()==col.ColumnName.ToLower())
                        {
                            tgtCols2.Add(col);
                            mappingTableSqlParameters.Add(srcFieldName, 
                                new SqlParameter("@" + col.ColumnName, DBNull.Value));
                            break;
                        }
                    }
                }
            }
            string spBody2 = StoredProcUtil.CreateStoredProcedureStatementForInsert(
                DataProviderType.MSSQL, spNameInsertMapping, string.Empty,
                this._Instruction.CollapseMappingTable,
                tgtCols2.ToArray());
            DbUtil.NewStoredProc(this._Instruction.TgtConnStr, spNameInsertMapping, spBody2);
            

            int startCollapseID = 0;
            string maxVal = DbUtil.GetTableFieldMaxValue(
                DataProviderType.MSSQL,
                this._Instruction.TgtConnStr,
                string.Empty,
                this._Instruction.CollapseInstanceTable,
                this._Instruction.CollapsedIdAttributeName);
            if (!string.IsNullOrEmpty(maxVal))
                startCollapseID = int.Parse(maxVal);

            try
            {
                if(totalEntityCount==0)
                    throw new Exception("Unable to read source data");

                this._AttributeComparers = new Dictionary<string, IAttributeDistanceCalcutor>();
                foreach (AttributeMetaData attrMetaData in this._Instruction.AttrMetaDatas)
                {
                    this._AttributeComparers.Add(attrMetaData.AttributeName, attrMetaData.GetComparer()); 
                }

                ICollapseCalculator collapseCalc = null;
                Assembly assembly = this.GetType().Assembly;
                Type type = assembly.GetType(this._Instruction.CollapseCalculatorTypeName);
                if (type != null)
                {
                    collapseCalc = (ICollapseCalculator)Activator.CreateInstance(type);
                }
                if (collapseCalc == null)
                    throw new Exception("Unable to find collapse calc class");

                srcConn1.Open();
                srcConn2.Open();
                IDbCommand srcCmd1 = srcConn1.CreateCommand();
                srcCmd1.CommandType = CommandType.Text;
                IDbCommand srcCmd2 = srcConn2.CreateCommand();
                srcCmd2.CommandType = CommandType.Text;

                tgtConn.Open();
                SqlCommand tgtCmd = tgtConn.CreateCommand();
                tgtCmd.CommandType = CommandType.StoredProcedure;

                if (this._Instruction.PartitionAttributes == null || this._Instruction.PartitionAttributes.Count == 0)
                    throw new Exception("Incorrect settings: must specify filter attributes");

                srcCmd1.CommandText = this._Instruction.SelectSql;
                srcCmd1.CommandTimeout = 0;
                IDataReader reader1 = srcCmd1.ExecuteReader();
                int entityProcessed = 0;
                int totalCollapsedEntityCount = 0;

                while(reader1.Read())
                {
                    string entityID = reader1[this._Instruction.IdAttributeName].ToString();
                    tgtCmd.CommandText =
                        string.Format("select count (*) from {0} where {1}={2}",
                                      this._Instruction.CollapseMappingTable,
                                      this._Instruction.CollapseMappingEntityIDField,
                                      FieldDataType.ToMssqlDataValue(entityIDDbType, entityID));
                    bool alreadyCollapsed = false;
                    try
                    {
                        tgtCmd.CommandType = CommandType.Text;
                        int existCount = int.Parse(tgtCmd.ExecuteScalar().ToString());
                        if (existCount > 0)
                            alreadyCollapsed = true;
                    }
                    catch
                    {
                        alreadyCollapsed = false;
                    }
                    if (alreadyCollapsed)
                    {
                        entityProcessed += 1;
                        if(this.WorkerStatusChangedEvent !=null && entityProcessed % 100==0)
                        {
                            string msg = string.Format("Processing {0} of {1} entities...", entityProcessed,
                                                       totalEntityCount);
                            int pct = entityProcessed*100/totalEntityCount;
                            this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction,
                                this.WorkflowName, this.ComponentName, msg, pct);
                        }
                        continue;
                    }

                    Dictionary<string, Entity> entities = new Dictionary<string, Entity>();
                    Entity entity1 = this.ReadEntityData(
                        reader1, this._Instruction.IdAttributeName, this._Instruction.AttrMetaDatas);
                    entities.Add(entityID, entity1);

                    foreach(EntityPartitionAttribute partAttr in this._Instruction.PartitionAttributes.Values)
                    {
                        AttributeMetaData correspondingMetaData = null;
                        foreach(AttributeMetaData metaData in this._Instruction.AttrMetaDatas)
                        {
                            if (metaData.AttributeName != null && metaData.AttributeName == partAttr.PartitionAttributeName)
                            {
                                correspondingMetaData = metaData;
                                break;
                            }
                        }

                        if(correspondingMetaData !=null)
                        {
                            object attrValue = entity1.ComparableAttributes[correspondingMetaData.AttributeName];
                            if(attrValue==null || attrValue.ToString().Length==0)
                                continue;

                            if(correspondingMetaData.AttributeType==AttributeType.VectorAttribute)
                            {
                                string[] attrValues = StringUtil.Split(attrValue.ToString());
                                if(attrValues !=null && attrValues.Length>0)
                                {
                                    for(int i=0;i<attrValues.Length;i++)
                                    {
                                        string attrValStr = attrValues[i];
                                        if (partAttr.IsExternal)
                                        {
                                            List<string> associatedEntityIDs =
                                                this.ReadEntityIDs(partAttr, attrValStr);
                                            if (associatedEntityIDs != null && associatedEntityIDs.Count > 0)
                                            {
                                                foreach (string associatedEntityID in associatedEntityIDs)
                                                {
                                                    if (!entities.ContainsKey(associatedEntityID))
                                                    {
                                                        srcCmd2.CommandText =
                                                            string.Format("select * from ({0}) as tmp where {1}={2}",
                                                                          this._Instruction.SelectSql,
                                                                          this._Instruction.IdAttributeName,
                                                                          associatedEntityID);
                                                        IDataReader srcReader2 = srcCmd2.ExecuteReader();
                                                        if (srcReader2.Read())
                                                        {
                                                            Entity associatedEntity = this.ReadEntityData(
                                                                srcReader2, this._Instruction.IdAttributeName,
                                                                this._Instruction.AttrMetaDatas);
                                                            if (!entities.ContainsKey(associatedEntity.Id))
                                                            {
                                                                entities.Add(associatedEntity.Id, associatedEntity);
                                                                if(entities.Count % 100==0 && this.WorkerStatusChangedEvent !=null)
                                                                {
                                                                    string msg =
                                                                        string.Format(
                                                                            "Collecting related entities for collapse...{0}",
                                                                            entities.Count);
                                                                    int pct = entityProcessed*100/totalEntityCount;
                                                                    this.WorkerStatusChangedEvent(this._JobTicketID,
                                                                                                  this.GetType(),
                                                                                                  this._Instruction,
                                                                                                  this.WorkflowName, this.ComponentName, 
                                                                                                  msg, pct);
                                                                }
                                                            }
                                                        }
                                                        srcReader2.Close();
                                                    }
                                                }
                                            }
                                        }
                                        else
                                        {
                                            {
                                                string filterSql1 = "(";
                                                if (this._Instruction.DbProvider == DataProviderType.MSSQL)
                                                {
                                                    attrValStr =
                                                        FieldDataType.ToMssqlDataValue(
                                                            srcFieldDbTypes[partAttr.PartitionAttributeName],
                                                            attrValStr);
                                                }
                                                else
                                                {
                                                    attrValStr =
                                                        FieldDataType.ToOracleDataValue(
                                                            srcFieldDbTypes[partAttr.PartitionAttributeName],
                                                            attrValStr);
                                                }
                                                filterSql1 += partAttr.PartitionAttributeName + "=" + attrValStr;
                                                filterSql1 = ")";

                                                srcCmd2.CommandText =
                                                    string.Format("select * from ({0}) as tmp where {1}",
                                                                  this._Instruction.SelectSql, filterSql1);
                                                IDataReader srcReader2 = srcCmd2.ExecuteReader();
                                                Dictionary<string, Entity> associatedEntities = this.ReadAllEntityData(
                                                    srcReader2, this._Instruction.IdAttributeName, this._Instruction.AttrMetaDatas);
                                                foreach (string entityID2 in associatedEntities.Keys)
                                                {
                                                    if (!entities.ContainsKey(entityID2))
                                                    {
                                                        entities.Add(entityID2, associatedEntities[entityID2]);
                                                        if (entities.Count % 100 == 0 && this.WorkerStatusChangedEvent != null)
                                                        {
                                                            string msg =
                                                                string.Format(
                                                                    "Collecting related entities for collapse...{0}",
                                                                    entities.Count);
                                                            int pct = entityProcessed * 100 / totalEntityCount;
                                                            this.WorkerStatusChangedEvent(this._JobTicketID,
                                                                                          this.GetType(),
                                                                                          this._Instruction,
                                                                                          this.WorkflowName, this.ComponentName, msg,
                                                                                          pct);
                                                        }
                                                    }
                                                }
                                                srcReader2.Close();
                                            }
                                        }
                                    }
                                }
                            }
                            else
                            {
                                string attrValStr = attrValue.ToString();
                                if (partAttr.IsExternal)
                                {
                                    List<string> associatedEntityIDs =
                                        this.ReadEntityIDs(partAttr, attrValue);
                                    if (associatedEntityIDs != null && associatedEntityIDs.Count > 0)
                                    {
                                        foreach (string associatedEntityID in associatedEntityIDs)
                                        {
                                            if (!entities.ContainsKey(associatedEntityID))
                                            {
                                                srcCmd2.CommandText =
                                                    string.Format("select * from ({0}) as tmp where {1}={2}",
                                                                  this._Instruction.SelectSql,
                                                                  this._Instruction.IdAttributeName,
                                                                  associatedEntityID);
                                                IDataReader srcReader2 = srcCmd2.ExecuteReader();
                                                if (srcReader2.Read())
                                                {
                                                    Entity associatedEntity = this.ReadEntityData(
                                                        srcReader2, this._Instruction.IdAttributeName,
                                                        this._Instruction.AttrMetaDatas);
                                                    if (!entities.ContainsKey(associatedEntity.Id))
                                                    {
                                                        entities.Add(associatedEntity.Id, associatedEntity);
                                                        if (entities.Count % 100 == 0 && this.WorkerStatusChangedEvent != null)
                                                        {
                                                            string msg =
                                                                string.Format(
                                                                    "Collecting related entities for collapse...{0}",
                                                                    entities.Count);
                                                            int pct = entityProcessed * 100 / totalEntityCount;
                                                            this.WorkerStatusChangedEvent(this._JobTicketID,
                                                                                          this.GetType(),
                                                                                          this._Instruction,
                                                                                          this.WorkflowName, this.ComponentName, msg,
                                                                                          pct);
                                                        }
                                                    }
                                                }
                                                srcReader2.Close();
                                            }
                                        }
                                    }
                                }
                                else
                                {
                                    {
                                        string filterSql1 = "(";
                                        if (this._Instruction.DbProvider == DataProviderType.MSSQL)
                                        {
                                            attrValStr =
                                                FieldDataType.ToMssqlDataValue(
                                                    srcFieldDbTypes[partAttr.PartitionAttributeName],
                                                    attrValStr);
                                        }
                                        else
                                        {
                                            attrValStr =
                                                FieldDataType.ToOracleDataValue(
                                                    srcFieldDbTypes[partAttr.PartitionAttributeName],
                                                    attrValStr);
                                        }
                                        filterSql1 += partAttr.PartitionAttributeName + "=" + attrValStr;
                                        filterSql1 += ")";
                                        string selectSql = this._Instruction.SelectSql;
                                        if(selectSql.ToLower().IndexOf("order by")>0)
                                        {
                                            selectSql = 
                                                selectSql.Substring(0, selectSql.ToLower().LastIndexOf("order by"));
                                        }
                                        srcCmd2.CommandText =
                                            string.Format("select * from ({0}) as tmp where {1}",
                                                          selectSql, filterSql1);
                                        IDataReader srcReader2 = srcCmd2.ExecuteReader();
                                        Dictionary<string, Entity> associatedEntities = this.ReadAllEntityData(
                                            srcReader2, this._Instruction.IdAttributeName, this._Instruction.AttrMetaDatas);
                                        foreach (string entityID2 in associatedEntities.Keys)
                                        {
                                            if (!entities.ContainsKey(entityID2))
                                            {
                                                entities.Add(entityID2, associatedEntities[entityID2]);
                                                if (entities.Count % 100 == 0 && this.WorkerStatusChangedEvent != null)
                                                {
                                                    string msg =
                                                        string.Format(
                                                            "Collecting related entities for collapse...{0}",
                                                            entities.Count);
                                                    int pct = entityProcessed * 100 / totalEntityCount;
                                                    this.WorkerStatusChangedEvent(this._JobTicketID,
                                                                                  this.GetType(),
                                                                                  this._Instruction,
                                                                                  this.WorkflowName, this.ComponentName, msg,
                                                                                  pct);
                                                }
                                            }
                                        }
                                        srcReader2.Close();
                                    }
                                }
                            }
                        }
                    }

                    if(this.WorkerStatusChangedEvent !=null)
                    {
                        string msg =
                            string.Format("Processed {0} of {1} entities: trying to collapse group of {2} entities",
                                          entityProcessed, totalEntityCount, entities.Count);
                        int pct = entityProcessed*100/totalEntityCount;
                        this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction,
                            this.WorkflowName, this.ComponentName, msg, pct);
                    }

                    Dictionary<int, Dictionary<string, double>> collapsedEntities =
                        new Dictionary<int, Dictionary<string, double>>();
                    Dictionary<int, Dictionary<string, object>> collapsedInstances =
                        new Dictionary<int, Dictionary<string, object>>();

                    if(entities.Count>2000)
                    {
                        Dictionary<int, Dictionary<string, Entity>> splittedEntities =
                            this.SplitEntities(entities, CountPerCollapseSplit);
                        Dictionary<int, Dictionary<int, Dictionary<string, double>>> collapsedEntitiesInGroups =
                            new Dictionary<int, Dictionary<int, Dictionary<string, double>>>();
                        Dictionary<int, Dictionary<int, Dictionary<string, object>>> collapsedInstancesInGroups =
                            new Dictionary<int, Dictionary<int, Dictionary<string, object>>>();

                        foreach(int splitID in splittedEntities.Keys)
                        {
                            if(this.WorkerStatusChangedEvent !=null)
                            {
                                string msg = string.Format("Trying to collapse splitted entities, {0} of {1}", splitID,
                                                           splittedEntities.Count);
                                int pct = 5;
                                this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction,
                                    this.WorkflowName, this.ComponentName, msg,
                                                              pct);
                            }

                            Dictionary<int, Dictionary<string, double>> splitCollapsedEntities =
                                new Dictionary<int, Dictionary<string, double>>();
                            Dictionary<int, Dictionary<string, object>> splitCollapsedInstances =
                                new Dictionary<int, Dictionary<string, object>>();
                            collapseCalc.TryCollapse(
                                this._Instruction.IdAttributeName,
                                this._Instruction.CollapsedIdAttributeName,
                                this._Instruction.AttrMetaDatas,
                                this._AttributeComparers,
                                splittedEntities[splitID],
                                this._Instruction.NativeAttributeThreshold,
                                this._Instruction.AssociativeAttributeThreshold,
                                this._Instruction.CollapseScoreThreshold,
                                ref splitCollapsedEntities,
                                ref splitCollapsedInstances);

                            Dictionary<int, Dictionary<string, double>> splitCollapsedEntitiesCopy =
                                new Dictionary<int, Dictionary<string, double>>();
                            Dictionary<int, Dictionary<string, object>> splitCollapsedInstancesCopy =
                                new Dictionary<int, Dictionary<string, object>>();
                            foreach(int splitCollapseID in splitCollapsedEntities.Keys)
                            {
                                startCollapseID++;
                                Dictionary<string, double> collapsedEntitiesInSplit =
                                    new Dictionary<string, double>();
                                Dictionary<string, object> collapsedInstanceInSplit =
                                    new Dictionary<string, object>();
                                foreach(string key in splitCollapsedEntities[splitCollapseID].Keys)
                                {
                                    collapsedEntitiesInSplit.Add(key,splitCollapsedEntities[splitCollapseID][key]);
                                }
                                foreach(string attrName in splitCollapsedInstances[splitCollapseID].Keys)
                                {
                                    collapsedInstanceInSplit.Add(attrName,splitCollapsedInstances[splitCollapseID][attrName]);
                                }
                                splitCollapsedEntitiesCopy.Add(startCollapseID, collapsedEntitiesInSplit);
                                splitCollapsedInstancesCopy.Add(startCollapseID, collapsedInstanceInSplit);
                            }

                            collapsedEntitiesInGroups.Add(splitID, splitCollapsedEntitiesCopy);
                            collapsedInstancesInGroups.Add(splitID, splitCollapsedInstancesCopy);
                        }

                        this.MergeCollapsedEntities(collapseCalc,
                            entities, collapsedEntitiesInGroups, collapsedInstancesInGroups,
                            ref collapsedEntities, ref collapsedInstances);
                    }
                    else
                    {
                        collapseCalc.TryCollapse(
                            this._Instruction.IdAttributeName,
                            this._Instruction.CollapsedIdAttributeName,
                            this._Instruction.AttrMetaDatas,
                            this._AttributeComparers,
                            entities,
                            this._Instruction.NativeAttributeThreshold,
                            this._Instruction.AssociativeAttributeThreshold,
                            this._Instruction.CollapseScoreThreshold,
                            ref collapsedEntities, ref collapsedInstances);
                        startCollapseID += collapsedEntities.Count;
                    }
                    

                    if(this.WorkerStatusChangedEvent !=null)
                    {
                        string msg = string.Format("Total of {0} collapses found from {1} entities",
                                                   collapsedInstances.Count, entities.Count);
                        int pct = entityProcessed * 100 / totalEntityCount;
                        this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction,
                            this.WorkflowName, this.ComponentName, msg, pct);
                    }

                    // merge
                    int[] collapseIDs = new int[collapsedEntities.Count];
                    collapsedEntities.Keys.CopyTo(collapseIDs, 0);
                    int entitiesMergedToExisting = 0;
                    int collapsesMergedExisting = 0;
                    for (int i = 0; i < collapseIDs.Length;i++)
                    {
                        int collapsedID = collapseIDs[i];
                        int mergeCollapseID = -1;
                        Dictionary<string, double> entityScores = collapsedEntities[collapsedID];
                        Dictionary<string, object> instance = collapsedInstances[collapsedID];
                        foreach (string entityID3 in entityScores.Keys)
                        {
                            tgtCmd.CommandText = string.Format(
                                "select max ({0}) from {1} where {2}={3}",
                                this._Instruction.CollapseMappingCollapseIDField,
                                this._Instruction.CollapseMappingTable,
                                this._Instruction.CollapseMappingEntityIDField,
                                FieldDataType.ToMssqlDataValue(entityIDDbType, entityID3));
                            try
                            {
                                mergeCollapseID = int.Parse(tgtCmd.ExecuteScalar().ToString());
                            }
                            catch
                            {
                                mergeCollapseID = -1;
                            }
                        }
                        if (mergeCollapseID > 0)
                        {
                            collapsesMergedExisting += 1;
                            entitiesMergedToExisting += collapsedEntities[collapsedID].Count;

                            this.MergeCollapseResult(
                                mergeCollapseID, tgtCmd,
                                spNameInsertInstance, entityIDDbType, srcFieldParameterMappings,
                                entities, entityScores, instance,
                                this._Instruction.CollapseMappingTable,
                                this._Instruction.CollapseMappingEntityIDField,
                                this._Instruction.CollapseMappingCollapseIDField,
                                this._Instruction.CollapseMappingScoreField,
                                this._Instruction.CollapseMappingTraceField);
                            collapsedEntities.Remove(collapsedID);
                            collapsedInstances.Remove(collapsedID);
                        }
                    }

                    if(entitiesMergedToExisting>0 && this.WorkerStatusChangedEvent !=null)
                    {
                        string msg =
                            string.Format("Merged {0} entities to {1} existing collapses",
                                          entitiesMergedToExisting, collapsesMergedExisting);
                        int pct = entityProcessed * 100 / totalEntityCount;
                        this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction,
                            this.WorkflowName, this.ComponentName, msg, pct);
                    }

                    this.SaveCollapseResult(
                        startCollapseID, tgtCmd, 
                        entities, collapsedEntities, collapsedInstances,
                        spNameInsertInstance, srcFieldParameterMappings,
                        this._Instruction.CollapseMappingEntityIDField,
                        this._Instruction.CollapseMappingCollapseIDField,
                        this._Instruction.CollapseMappingScoreField,
                        this._Instruction.CollapseMappingTraceField,
                        spNameInsertMapping, mappingTableSqlParameters);

                    totalCollapsedEntityCount += collapsedEntities.Count;
                    foreach(int collapseID in collapsedEntities.Keys)
                    {
                        entityProcessed += collapsedEntities[collapseID].Count;    
                    }
                    

                    if(this.WorkerStatusChangedEvent !=null)
                    {
                        string msg = string.Format("Collapse: processed {0} of {1} entities",
                                                   entityProcessed, totalEntityCount);
                        int pct = entityProcessed * 100 / totalEntityCount;
                        this.WorkerStatusChangedEvent(
                            this._JobTicketID, this.GetType(),
                            this._Instruction, this.WorkflowName, this.ComponentName, msg, pct);
                    }
                }
                reader1.Close();

                this._Output =
                    new StreamingCollapseOutput(this._Instruction.OwnerComponentID,
                                       this._JobTicketID, totalEntityCount, 
                                       totalCollapsedEntityCount,
                                       this._Instruction.CollapseResultFilePath,
                                       this._Instruction.CollapseInstanceTable,
                                       this._Instruction.CollapseMappingTable);

                if(this.WorkerFinishedEvent !=null)
                {
                    this.WorkerFinishedEvent(this._JobTicketID, this.GetType(), this._Instruction, this._Output);
                }
            }
            catch(Exception ex)
            {
                if(this.WorkerErredEvent !=null)
                {
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.StackTrace,true);
                }
            }
            finally
            {
                srcConn1.Close();
                srcConn2.Close();
                tgtConn.Close();
            }
        }
        #endregion

        #region reading data 
        private Entity ReadEntityData(IDataReader reader, string idAttrName, List<AttributeMetaData> metaDatas)
        {
            string entityID = reader[idAttrName].ToString();
            Dictionary<string, object> attrVals = new Dictionary<string, object>();
            foreach (AttributeMetaData attrMetaData in metaDatas)
            {
                if (attrMetaData.IsExternalAttribute)
                {
                    if (attrMetaData.AttributeType == AttributeType.VectorAttribute)
                    {
                        List<object> objs = this.GetExternalAttributeVectorData(
                            attrMetaData.ExternalAttributeDbProvider,
                            attrMetaData.ExternalAttributeConnStr,
                            attrMetaData.ExternalAttributeSelectSql,
                            attrMetaData.AttributeName,
                            entityID);
                        List<string> objStrVals = new List<string>();
                        if (objs != null && objs.Count > 0)
                        {
                            foreach (object obj in objs)
                            {
                                objStrVals.Add(obj.ToString());
                            }
                        }
                        string objStrVal = StringUtil.ToString(objStrVals.ToArray());
                        attrVals.Add(attrMetaData.AttributeName, objStrVal);
                    }
                    else
                    {
                        object obj = this.GetExternalAttributeData(
                            attrMetaData.ExternalAttributeDbProvider,
                            attrMetaData.ExternalAttributeConnStr,
                            attrMetaData.ExternalAttributeSelectSql,
                            attrMetaData.AttributeName,
                            entityID);
                        attrVals.Add(attrMetaData.AttributeName, obj);
                    }
                }
                else if (attrMetaData.AttributeType == AttributeType.CompositeAttribute)
                {
                    foreach (string attrName in attrMetaData.ArgNameAttrNameBindings.Values)
                    {
                        attrVals.Add(attrName, reader[attrName]);
                    }
                }
                else
                {
                    attrVals.Add(attrMetaData.AttributeName, reader[attrMetaData.AttributeName]);
                }
            }
            Entity entity=new Entity(entityID, attrVals);
            return entity;
        }

        private Dictionary<string, Entity> ReadAllEntityData(IDataReader reader, string idAttrName, List<AttributeMetaData> metaDatas)
        {
            Dictionary<string, Entity> entities = new Dictionary<string, Entity>();
            while (reader.Read())
            {
                Entity entity = this.ReadEntityData(reader, idAttrName, metaDatas);
                if (entity != null && !entities.ContainsKey(entity.Id))
                    entities.Add(entity.Id, entity);
            }
            return entities;
        }

        private object GetExternalAttributeData(
            DataProviderType dbProvider,
            string connStr,
            string selectSql,
            string extAttrName,
            string entityID)
        {
            object attrValue = null;
            IDbConnection conn = null;
            if (dbProvider == DataProviderType.MSSQL)
                conn = new SqlConnection(connStr);
            else
                conn = new OracleConnection(connStr);
            try
            {
                conn.Open();
                IDbCommand cmd = conn.CreateCommand();
                cmd.CommandType = CommandType.Text;
                if (selectSql.IndexOf("{0}") > 0)
                    selectSql = selectSql.Replace("{0}", entityID.ToString());
                cmd.CommandText = selectSql;
                IDataReader reader = cmd.ExecuteReader();
                if(reader.Read())
                {
                    attrValue = reader[extAttrName];
                }
                reader.Close();
            }
            catch (Exception ex)
            {
                if (this.WorkerErredEvent != null)
                {
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.StackTrace,true);
                }

            }
            finally
            {
                conn.Close();
            }
            return attrValue;
        }

        private List<object> GetExternalAttributeVectorData(
            DataProviderType dbProvider,
            string connStr,
            string selectSql,
            string extAttrName,
            string entityID)
        {
            List<object> objs=new List<object>();
            IDbConnection conn = null;
            if (dbProvider == DataProviderType.MSSQL)
                conn = new SqlConnection(connStr);
            else
                conn = new OracleConnection(connStr);
            try
            {
                conn.Open();
                IDbCommand cmd = conn.CreateCommand();
                cmd.CommandType = CommandType.Text;
                if (selectSql.IndexOf("{0}") > 0)
                    selectSql = selectSql.Replace("{0}", entityID.ToString());
                cmd.CommandText = selectSql;
                IDataReader reader = cmd.ExecuteReader();
                while (reader.Read())
                {
                    object attrValue = null;
                    attrValue = reader[extAttrName];
                    objs.Add(attrValue);
                }
                reader.Close();
            }
            catch (Exception ex)
            {
                if (this.WorkerErredEvent != null)
                {
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.StackTrace,true);
                }
            }
            finally
            {
                conn.Close();
            }

            return objs;
        }

        private List<string> ReadEntityIDs(
            EntityPartitionAttribute partitionAttr, 
            object attrValue)
        {
            List<string> entityIDs = new List<string>();

            if(attrValue==null)
                return entityIDs;

            
            IDbConnection conn = null;
            if (partitionAttr.DbProvider == DataProviderType.MSSQL)
                conn = new SqlConnection(partitionAttr.ConnStr);
            else
                conn = new OracleConnection(partitionAttr.ConnStr);
            try
            {
                conn.Open();
                IDbCommand cmd = conn.CreateCommand();
                cmd.CommandType = CommandType.Text;
                string selectSql = partitionAttr.SelectSql;
                if (selectSql.IndexOf("{0}") > 0)
                {
                    string attrValString = FieldDataType.ToMssqlDataValue(partitionAttr.PartitionAttributeDbType, attrValue.ToString());
                    if (partitionAttr.DbProvider == DataProviderType.Oracle)
                        attrValString = FieldDataType.ToOracleDataValue(partitionAttr.PartitionAttributeDbType, attrValue.ToString());
                    selectSql = selectSql.Replace("{0}", attrValString);
                }
                cmd.CommandText = selectSql;
                IDataReader reader = cmd.ExecuteReader();
                while (reader.Read())
                {
                    string entityID = reader[partitionAttr.EntityIDFieldName].ToString();
                    entityIDs.Add(entityID);
                }
                reader.Close();
            }
            catch (Exception ex)
            {
                if (this.WorkerErredEvent != null)
                {
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.StackTrace,true);
                }
            }
            finally
            {
                conn.Close();
            }

            return entityIDs;
        }
        #endregion

        #region save
        private void SaveCollapseResult(
            int startCollapseID,
            SqlCommand tgtCmd,
            Dictionary<string, Entity> entities,
            Dictionary<int, Dictionary<string, double>> collapsedEntities,
            Dictionary<int, Dictionary<string, object>> collapsedInstances,
            string spNameInsertInstance, 
            Dictionary<string, SqlParameter> instanceTableSqlParameters,
            string mappingEntityIDField,
            string mappingCollapseIDField, 
            string mappingScoreField,
            string mappingTraceField,
            string spNameInsertMapping,
            Dictionary<string,SqlParameter> mappingTableSqlParameters)
        {
            if (tgtCmd.Connection.State != ConnectionState.Open)
                tgtCmd.Connection.Open();

            tgtCmd.CommandText = spNameInsertInstance;
            tgtCmd.CommandType = CommandType.StoredProcedure;

            TransactionOptions options = new TransactionOptions();
            options.IsolationLevel = System.Transactions.IsolationLevel.ReadCommitted;
            options.Timeout = new TimeSpan(0, 10, 0, 0);     // 10 hours
            TransactionScope scope = new TransactionScope(TransactionScopeOption.RequiresNew, options);
            try
            {
                int instanceInserted = 0;
                foreach (int collapsedID in collapsedInstances.Keys)
                {
                    int newCollapseID = startCollapseID + collapsedID;
                    Dictionary<string, object> instanceAttrValues = collapsedInstances[collapsedID];
                    tgtCmd.Parameters.Clear();
                    SqlParameter collapsedIdParameter =
                        instanceTableSqlParameters[this._Instruction.CollapsedIdAttributeName];
                    collapsedIdParameter.Value = newCollapseID;
                    tgtCmd.Parameters.Add(collapsedIdParameter);
                    foreach (string srcFieldName in instanceTableSqlParameters.Keys)
                    {
                        SqlParameter parameter =
                            instanceTableSqlParameters[srcFieldName];
                        if (srcFieldName != this._Instruction.CollapsedIdAttributeName)
                        {
                            if (instanceAttrValues[srcFieldName] != null)
                                parameter.Value = instanceAttrValues[srcFieldName];
                            else
                                parameter.Value = DBNull.Value;

                            tgtCmd.Parameters.Add(parameter);
                        }
                    }

                    tgtCmd.ExecuteNonQuery();

                    instanceInserted++;
                    if (this.WorkerStatusChangedEvent != null && instanceInserted % 100 == 0)
                    {
                        string msg = string.Format("Saving {0} of {1} collapsed instances to database",
                                                   instanceInserted, collapsedInstances.Count);
                        int pct = instanceInserted * 100 / collapsedInstances.Count;
                        this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction,
                            this.WorkflowName, this.ComponentName, msg, pct);
                    }
                }

                tgtCmd.CommandType = CommandType.StoredProcedure;
                tgtCmd.CommandText = spNameInsertMapping;
                int entitySaved = 0;
                int totalEntityCount = 0;
                foreach (int collapseID in collapsedEntities.Keys)
                {
                    totalEntityCount += collapsedEntities[collapseID].Count;
                }

                foreach (int collapseID in collapsedEntities.Keys)
                {
                    int newCollapseID = startCollapseID + collapseID;
                    Dictionary<string, double> entityScores = collapsedEntities[collapseID];
                    foreach (string entityID in entityScores.Keys)
                    {
                        double score = entityScores[entityID];
                        Entity entity = entities[entityID];

                        tgtCmd.Parameters.Clear();
                        foreach(string colName in mappingTableSqlParameters.Keys)
                        {
                            if(colName==mappingCollapseIDField)
                            {
                                mappingTableSqlParameters[mappingCollapseIDField].Value = newCollapseID;                                
                            }
                            else if(colName==mappingEntityIDField)
                            {
                                mappingTableSqlParameters[mappingEntityIDField].Value = entityID;
                            }
                            else if(colName==mappingScoreField)
                            {
                                mappingTableSqlParameters[mappingScoreField].Value = score;
                            }
                            else if(colName==mappingTraceField)
                            {
                                string trace = entity.Trace.Replace("'", "''");
                                if (trace.Length > 1900)
                                    trace = trace.Substring(0, 1900) + "...";
                                mappingTableSqlParameters[mappingTraceField].Value = trace;
                            }
                            else if (entity.ComparableAttributes.ContainsKey(colName))
                            {
                                if(entity.ComparableAttributes[colName]!=null)
                                {
                                    mappingTableSqlParameters[colName].Value = entity.ComparableAttributes[colName];
                                }
                                else
                                {
                                    mappingTableSqlParameters[colName].Value = DBNull.Value;
                                }
                            }
                            else
                            {
                                mappingTableSqlParameters[colName].Value = DBNull.Value;
                            }

                            tgtCmd.Parameters.Add(mappingTableSqlParameters[colName]);
                        }
                        
                        tgtCmd.ExecuteNonQuery();

                        entitySaved++;
                        if (this.WorkerStatusChangedEvent != null && entitySaved % 100 == 0)
                        {
                            string msg = string.Format("Saving collapsed entity scores, {0} of {1} saved...",
                                                       entitySaved, totalEntityCount);
                            int pct = entitySaved * 100 / totalEntityCount;
                            this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction,
                                this.WorkflowName, this.ComponentName, msg, pct);
                        }
                    }
                }

                if (this.WorkerStatusChangedEvent != null && entitySaved % 100 == 0)
                {
                    string msg = string.Format("Saving collapsed entity scores, {0} of {1} saved...",
                                               entitySaved, totalEntityCount);
                    int pct = 100;
                    this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction,
                        this.WorkflowName, this.ComponentName, msg, pct);
                }

                scope.Complete();
            }
            catch (Exception ex)
            {
                if (this.WorkerErredEvent != null)
                {
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.StackTrace,true);
                }
            }
            finally
            {
                scope.Dispose();
            }
        }

        private void MergeCollapseResult(
            int mergeCollapseID,
            SqlCommand tgtCmd, string spName, SqlDbType entityIDDbType,
            Dictionary<string, SqlParameter> fieldNameParameterMappings,
            Dictionary<string, Entity> entities,
            Dictionary<string, double> entityScores,
            Dictionary<string, object> collapsedInstance,
            string mappingTable, string mappingEntityIDField,
            string mappingCollapseIDField, 
            string mappingScoreField,
            string mappingTraceField)
        {
            TransactionOptions options = new TransactionOptions();
            options.IsolationLevel = System.Transactions.IsolationLevel.ReadCommitted;
            options.Timeout = new TimeSpan(0, 10, 0, 0);     // 10 hours
            TransactionScope scope = new TransactionScope(TransactionScopeOption.RequiresNew, options);
            try
            {
                if (tgtCmd.Connection.State != ConnectionState.Open)
                    tgtCmd.Connection.Open();
                tgtCmd.CommandText = spName;
                tgtCmd.CommandType = CommandType.StoredProcedure;

                tgtCmd.Parameters.Clear();
                foreach (string srcFieldName in fieldNameParameterMappings.Keys)
                {
                    SqlParameter parameter =
                        fieldNameParameterMappings[srcFieldName];
                    if (srcFieldName == this._Instruction.CollapsedIdAttributeName)
                        parameter.Value = mergeCollapseID;
                    else if (collapsedInstance.ContainsKey(srcFieldName) &&
                        collapsedInstance[srcFieldName] != null)
                        parameter.Value = collapsedInstance[srcFieldName];
                    else
                        parameter.Value = DBNull.Value;
                    tgtCmd.Parameters.Add(parameter);
                }
                tgtCmd.ExecuteNonQuery();

                tgtCmd.CommandType = CommandType.Text;
                foreach (string entityID in entityScores.Keys)
                {
                    double score = entityScores[entityID];
                    string trace = entities[entityID].Trace;
                    trace = trace.Replace("'", "''");
                    if (trace.Length > 1900)
                        trace = trace.Substring(0, 1900) + "...";
                    tgtCmd.CommandText =
                        string.Format(
                            "insert into {0} ({1},{2},{3},{4}) values " +
                            "({5},{6}, {7},'{8}')",
                            mappingTable, mappingCollapseIDField, mappingEntityIDField,
                            mappingScoreField,
                            mappingTraceField,
                            mergeCollapseID, 
                            FieldDataType.ToMssqlDataValue(entityIDDbType, entityID), 
                            score, trace);
                    tgtCmd.ExecuteNonQuery();
                }

                scope.Complete();
            }
            catch (Exception ex)
            {
                if(this.WorkerErredEvent !=null)
                {
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.StackTrace,true);
                }
            }
            finally
            {
                scope.Dispose();
            }
        }
        #endregion

        #region split and merge
        private Dictionary<int,Dictionary<string,Entity>> SplitEntities(
            Dictionary<string,Entity> entities,
            int maxCountInGroup)
        {
            Dictionary<int,Dictionary<string,Entity>> splittedEntities=new Dictionary<int, Dictionary<string, Entity>>();
            Dictionary<string,Entity> currentSplit=new Dictionary<string, Entity>();
            foreach(string key in entities.Keys)
            {
                if (currentSplit.Count < maxCountInGroup)
                {
                    currentSplit.Add(key, entities[key]);
                }
                else
                {
                    int newSplitID = splittedEntities.Count + 1;
                    splittedEntities.Add(newSplitID, currentSplit);
                    currentSplit = new Dictionary<string, Entity>();
                    currentSplit.Add(key, entities[key]);
                }
            }
            if(currentSplit.Count>0)
            {
                int newSplitID = splittedEntities.Count + 1;
                splittedEntities.Add(newSplitID, currentSplit);
            }
            return splittedEntities;
        }

        private void MergeCollapsedEntities(
            ICollapseCalculator collapseCalc,
            Dictionary<string,Entity> entities, 
            Dictionary<int,Dictionary<int,Dictionary<string,double>>> splittedCollapsedEntities,
            Dictionary<int,Dictionary<int, Dictionary<string,object>>> splittedCollapsedInstances,
            ref Dictionary<int,Dictionary<string,double>> mergedCollapsedEntities,
            ref Dictionary<int,Dictionary<string,object>> mergedCollapsedInstances)
        {
            mergedCollapsedEntities=new Dictionary<int, Dictionary<string, double>>();
            mergedCollapsedInstances=new Dictionary<int, Dictionary<string, object>>();

            Dictionary<int, int> collapseID_splitID_Mappings = new Dictionary<int, int>();
            foreach(int splitID in splittedCollapsedInstances.Keys)
            {
                Dictionary<int, Dictionary<string, object>> instances = splittedCollapsedInstances[splitID];
                foreach(int partitionID in instances.Keys)
                {
                    collapseID_splitID_Mappings.Add(partitionID,splitID);
                }
            }

            Dictionary<int, List<int>> mergeableCollapseIDs = new Dictionary<int, List<int>>();
            int[] splitIDs=new int[splittedCollapsedEntities.Count];
            splittedCollapsedEntities.Keys.CopyTo(splitIDs,0);

            Dictionary<string, AttributeMetaData> attrMetaDataByName=new Dictionary<string, AttributeMetaData>();
            foreach(AttributeMetaData attrMetaData in this._Instruction.AttrMetaDatas)
            {
                if (!attrMetaDataByName.ContainsKey(attrMetaData.AttributeName))
                    attrMetaDataByName.Add(attrMetaData.AttributeName, attrMetaData);
            }

            for(int i=0;i<splitIDs.Length;i++)
            {
                if(this.WorkerStatusChangedEvent !=null)
                {
                    string msg = string.Format(
                        "Trying to merge splited collapsed entities, {0} of {1}",
                        (i + 1), splitIDs.Length);
                    int pct = 75;
                    this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction,
                        this.WorkflowName, this.ComponentName, msg, pct);
                }

                Dictionary<int, Dictionary<string, double>> collapsedEntities1 =
                    splittedCollapsedEntities[splitIDs[i]];

                for(int k=i+1;k<splitIDs.Length;k++)
                {
                    Dictionary<int, Dictionary<string, double>> collapsedEntities2 =
                        splittedCollapsedEntities[splitIDs[k]];
                    Dictionary<int, int> mergedCollapseIDPairs = new Dictionary<int, int>();
                    bool foundMerge = this.CanMerge(
                        collapseCalc, attrMetaDataByName, this._AttributeComparers,
                        collapsedEntities1, collapsedEntities2,
                        entities, ref mergedCollapseIDPairs);
                    if(foundMerge)
                    {
                        foreach(int fromMergeID in mergedCollapseIDPairs.Keys)
                        {
                            int toMergeID = mergedCollapseIDPairs[fromMergeID];
                            int mergeCollapseID1 = Math.Min(fromMergeID, toMergeID);
                            int mergeCollapseID2 = Math.Max(fromMergeID, toMergeID);
                            if (mergeableCollapseIDs.ContainsKey(mergeCollapseID1))
                            {
                                List<int> mergeCollapseIDs2 = mergeableCollapseIDs[mergeCollapseID1];
                                if (!mergeCollapseIDs2.Contains(mergeCollapseID2))
                                    mergeCollapseIDs2.Add(mergeCollapseID2);
                                mergeableCollapseIDs[mergeCollapseID1] = mergeCollapseIDs2;
                            }
                            else
                            {
                                List<int> mergeCollapseIDs2 = new List<int>();
                                mergeCollapseIDs2.Add(mergeCollapseID2);
                                mergeableCollapseIDs.Add(mergeCollapseID1, mergeCollapseIDs2);
                            }
                        }
                        
                    }
                }
            }
            
            List<int> addedCollapseIDs=new List<int>();
            foreach(int splitID in splittedCollapsedEntities.Keys)
            {
                if(this.WorkerStatusChangedEvent !=null)
                {
                    string msg = string.Format("Trying to determine associated entities for split {0} of {1}",
                                               splitID, splittedCollapsedEntities.Count);
                    int pct = splitID*100/splittedCollapsedEntities.Count;
                    this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction,
                        this.WorkflowName, this.ComponentName, msg, pct);
                }
                Dictionary<int, Dictionary<string, double>> collapsedEntities1 = splittedCollapsedEntities[splitID];
                foreach(int collapseID in collapsedEntities1.Keys)
                {
                    if (!addedCollapseIDs.Contains(collapseID))
                    {
                        List<int> collapseIDsToMerge = new List<int>();
                        this.GetAssociatedCollapseIDs(collapseID, mergeableCollapseIDs, ref collapseIDsToMerge);
                        //collapseIDsToMerge.Add(collapseID);
                        //if (mergeableCollapseIDs.ContainsKey(collapseID))
                        //{
                        //    foreach (int collapseIDToMerge in mergeableCollapseIDs[collapseID])
                        //    {
                        //        if (!addedCollapseIDs.Contains(collapseIDToMerge) &&
                        //            !collapseIDsToMerge.Contains(collapseIDToMerge))
                        //        {
                        //            collapseIDsToMerge.Add(collapseIDToMerge);
                        //        }
                        //    }
                        //}
                        //else
                        //{
                        //    foreach (int fromCollapseID in mergeableCollapseIDs.Keys)
                        //    {
                        //        if (mergeableCollapseIDs[fromCollapseID].Contains(collapseID))
                        //        {
                        //            if (!addedCollapseIDs.Contains(fromCollapseID) &&
                        //                !collapseIDsToMerge.Contains(fromCollapseID))
                        //            {
                        //                collapseIDsToMerge.Add(fromCollapseID);
                        //                foreach (int toCollapseID in mergeableCollapseIDs[fromCollapseID])
                        //                {
                        //                    if (!addedCollapseIDs.Contains(toCollapseID) &&
                        //                        !collapseIDsToMerge.Contains(toCollapseID))
                        //                    {
                        //                        collapseIDsToMerge.Add(toCollapseID);
                        //                    }
                        //                }
                        //            }
                        //        }
                        //    }
                        //}

                        int commonCollapseID = IntUtil.Minimum(collapseIDsToMerge);
                        Dictionary<string, double> collapsedEntities = new Dictionary<string, double>();
                        foreach (int collapseID3 in collapseIDsToMerge)
                        {
                            if (!addedCollapseIDs.Contains(collapseID3))
                                addedCollapseIDs.Add(collapseID3);
                            int containingSplitID = collapseID_splitID_Mappings[collapseID3];
                            Dictionary<string, double> entityScores =
                                splittedCollapsedEntities[containingSplitID][collapseID3];
                            foreach (string key in entityScores.Keys)
                            {
                                collapsedEntities.Add(key, entityScores[key]);
                            }
                        }
                        mergedCollapsedEntities.Add(commonCollapseID, collapsedEntities);
                        int splitID2 = collapseID_splitID_Mappings[commonCollapseID];
                        mergedCollapsedInstances.Add(commonCollapseID,
                                                     splittedCollapsedInstances[splitID2][commonCollapseID]);
                    }
                }
            }
        }

        private bool CanMerge(
            ICollapseCalculator collapseCalc,
            Dictionary<string,AttributeMetaData> attrMetaDataByName,
            Dictionary<string, IAttributeDistanceCalcutor> attrComparers, 
            Dictionary<int,Dictionary<string,double>> collapsedEntities1,
            Dictionary<int,Dictionary<string,double>> collapsedEntities2,
            Dictionary<string,Entity> entities,
            ref Dictionary<int,int> mergedCollapseIDPairs)
        {
            int[] collapseIDs1=new int[collapsedEntities1.Count];
            collapsedEntities1.Keys.CopyTo(collapseIDs1, 0);
            int[] collapseIDs2=new int[collapsedEntities2.Count];
            collapsedEntities2.Keys.CopyTo(collapseIDs2, 0);
            foreach(int collapseID1 in collapseIDs1)
            {
                Dictionary<string, double> entities1 = collapsedEntities1[collapseID1];
                foreach(int collapseID2 in collapseIDs2)
                {
                    bool foundMergeableCollapse = false;
                    Dictionary<string, double> entities2 = collapsedEntities2[collapseID2];
                    foreach(string entityID1 in entities1.Keys)
                    {
                        Entity entity1 = entities[entityID1];
                        foreach(string entityID2 in entities2.Keys)
                        {
                            Entity entity2 = entities[entityID2];
                            
                            if(this._Instruction.MergeableAttributeNames !=null && this._Instruction.MergeableAttributeNames.Count>0)
                            {
                                if (collapseCalc.CanMerge(entity1, entity2, attrMetaDataByName, attrComparers, this._Instruction.MergeableAttributeNames))
                                {
                                    foundMergeableCollapse = true;
                                    if (!mergedCollapseIDPairs.ContainsKey(collapseID1))
                                    {
                                        mergedCollapseIDPairs.Add(collapseID1, collapseID2);
                                    }
                                    else if (mergedCollapseIDPairs[collapseID1] != collapseID2)
                                    {
                                        int existingCollapseID2 = mergedCollapseIDPairs[collapseID1];
                                        if (existingCollapseID2 < collapseID2 && !mergedCollapseIDPairs.ContainsKey(existingCollapseID2))
                                        {
                                            mergedCollapseIDPairs.Add(existingCollapseID2, collapseID2);
                                        }
                                        else if (collapseID2 < existingCollapseID2 && !mergedCollapseIDPairs.ContainsKey(collapseID2))
                                        {
                                            mergedCollapseIDPairs.Add(collapseID2, existingCollapseID2);
                                        }
                                    }
                                    break;
                                }
                            }
                            else
                            {
                                double similarityScore = 0;
                                if (collapseCalc.CanCollapse(
                                    this._Instruction.AttrMetaDatas,
                                    this._AttributeComparers,
                                    this._Instruction.NativeAttributeThreshold,
                                    this._Instruction.AssociativeAttributeThreshold,
                                    this._Instruction.CollapseScoreThreshold,
                                    entity1, entity2, ref similarityScore))
                                {
                                    foundMergeableCollapse = true;
                                    if (!mergedCollapseIDPairs.ContainsKey(collapseID1))
                                    {
                                        mergedCollapseIDPairs.Add(collapseID1, collapseID2);
                                    }
                                    else if (mergedCollapseIDPairs[collapseID1] != collapseID2)
                                    {
                                        int existingCollapseID2 = mergedCollapseIDPairs[collapseID1];
                                        if (existingCollapseID2 < collapseID2 && !mergedCollapseIDPairs.ContainsKey(existingCollapseID2))
                                        {
                                            mergedCollapseIDPairs.Add(existingCollapseID2, collapseID2);
                                        }
                                        else if (collapseID2 < existingCollapseID2 && !mergedCollapseIDPairs.ContainsKey(collapseID2))
                                        {
                                            mergedCollapseIDPairs.Add(collapseID2, existingCollapseID2);
                                        }
                                    }
                                    break;
                                }
                            }
                        }
                    }
                    if(foundMergeableCollapse)
                        break;
                }
            }

            if (mergedCollapseIDPairs.Count > 0)
                return true;
            else
                return false;
        }

        private void GetAssociatedCollapseIDs(
            int seedCollapseID, 
            Dictionary<int,List<int>> mergeableCollapseIDs,
            ref List<int> associatedCollapseIDs)
        {
            if (!associatedCollapseIDs.Contains(seedCollapseID))
                associatedCollapseIDs.Add(seedCollapseID);

            List<int> newlyAddedIDs=new List<int>();
            if(mergeableCollapseIDs.ContainsKey(seedCollapseID))
            {
                List<int> collapseIDs2 = mergeableCollapseIDs[seedCollapseID];
                foreach(int collapseID2 in collapseIDs2)
                {
                    if(!associatedCollapseIDs.Contains(collapseID2))
                    {
                        newlyAddedIDs.Add(collapseID2);
                        associatedCollapseIDs.Add(collapseID2);
                    }
                }
            }
            foreach(int fromID in mergeableCollapseIDs.Keys)
            {
                List<int> toIDs = mergeableCollapseIDs[fromID];
                if(IntUtil.Intercept(toIDs, associatedCollapseIDs))
                {
                    if(!associatedCollapseIDs.Contains(fromID))
                    {
                        newlyAddedIDs.Add(fromID);
                        associatedCollapseIDs.Add(fromID);
                    }
                    foreach(int toID in toIDs)
                    {
                        if(!associatedCollapseIDs.Contains(toID))
                        {
                            newlyAddedIDs.Add(toID);
                            associatedCollapseIDs.Add(toID);
                        }
                    }
                }
            }

            if(newlyAddedIDs.Count>0)
            {
                foreach(int newID in newlyAddedIDs)
                {
                    this.GetAssociatedCollapseIDs(newID, mergeableCollapseIDs, ref associatedCollapseIDs);
                }
            }
        }
        #endregion
    }
}
